package cn.doitedu.rtdw.data_etl;

import cn.doitedu.rtdw.data_sync.Gps2GeoHashUdf;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job1_用户行为日志公共维度退维 {

    public static void main(String[] args) {

        // 创建编程入口
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 创建如下逻辑表：
        // 1. kafka中的用户行为日志的映射表  ==> 中心事实表

        // 2. hbase中的注册信息维表   |
        // 3. hbase中的地理位置维表   |==> 要用到的维表
        // 4. hbase中的页面信息维表   |

        // 5. kafka中明细层数据topic 映射表   |
        // 6. doris中明细层数据 映射表        |=>  结果表

        // 1.
        // {"release_channel":"360应用市场",
        // "device_type":"mi6",
        // "session_id":"s01",
        // "lat":38.089969323508726,
        // "lng":114.35731900345093,
        // "username":"tiger",
        // "eventId":"add_cart",
        // "eventTime":1670583693000,
        // "properties":{"url":"/content/article/2354.html?a=3","itemId":"item002"}}
        tenv.executeSql(
                " CREATE TABLE events_source (                               "
                        + "     username     string,                                   "
                        + "     session_id   string,                                   "
                        + "     eventId      string,                                   "
                        + "     eventTime    bigint,                                   "
                        + "     lat          double ,                                  "
                        + "     lng          double ,                                  "
                        + "     release_channel   string,                              "
                        + "     device_type       string,                              "
                        + "     properties   map<string,string>,                       "
                        + "     proc_time AS proctime()                                "
                        + " ) WITH (                                                   "
                        + "  'connector' = 'kafka',                                    "
                        + "  'topic' = 'ods_events',                                   "
                        + "  'properties.bootstrap.servers' = 'doitedu:9092',          "
                        + "  'properties.group.id' = 'goo1',                           "
                        + "  'scan.startup.mode' = 'latest-offset',                    "
                        + "  'value.format'='json',                                    "
                        + "  'value.json.fail-on-missing-field'='false',               "
                        + "  'value.fields-include' = 'EXCEPT_KEY'                     "
                        + " )                                                          ");

        // 2. hbase中的用户注册信息维表
        tenv.executeSql(
                " create table dim_user_info_hbase(                                        "+
                        "    username string,                                                    "+
                        " 	 f1 row<id bigint,member_level_id bigint,phone string,status int,       "+
                        " 	       create_time timestamp(3), gender int,birthday date,city string, "+
                        " 		   job string,source_type int,modify_time timestamp(3)>            "+
                        " ) with (                                                                 "+
                        "  'connector' = 'hbase-2.2',                                              "+
                        "  'table-name' = 'dim_user_info',                                         "+
                        "  'zookeeper.quorum' = 'doitedu:2181'                                     "+
                        " )                                                                        "
        );

        // 3.hbase中的地理位置维表
        tenv.executeSql(
                " create table dim_geo_area_hbase(                    "+
                        "   geohash string,                                    "+
                        " 	f row<province string,city string,region string>,  "+
                        "   primary key(geohash) not enforced      "+
                        " ) with (                                 "+
                        "     'connector' = 'hbase-2.2',           "+
                        "     'table-name' = 'dim_geo_area',       "+
                        "     'zookeeper.quorum' = 'doitedu:2181'  "+
                        " )                                         "
        );

        // 4.hbase中的页面信息维表
        tenv.executeSql(
                " create table dim_page_info_hbase(              "+
                        "   url_prefix string,                            "+
                        " 	f row<page_service string,page_type string>,  "+
                        "   primary key(url_prefix) not enforced          "+
                        " ) with (                                        "+
                        "     'connector' = 'hbase-2.2',                  "+
                        "     'table-name' = 'dim_page_info',             "+
                        "     'zookeeper.quorum' = 'doitedu:2181'         "+
                        " )                                               "
        );


        // 5.建表，映射 kafka中用于存储维度打宽明细数据的topic
        tenv.executeSql(
                "  CREATE TABLE dwd_kafka(                                "
                        +"     user_id           BIGINT,                     "
                        +"     username          string,                     "
                        +"     session_id        string,                     "
                        +"     event_Id          string,                     "
                        +"     event_time        bigint,                     "
                        +"     lat               double,                     "
                        +"     lng               double,                     "
                        +"     release_channel   string,                     "
                        +"     device_type       string,                     "
                        +"     properties        map<string,string>,         "
                        +"     register_phone    STRING,                     "
                        +"     user_status       INT,                        "
                        +"     register_time     TIMESTAMP(3),               "
                        +"     register_gender   INT,                        "
                        +"     register_birthday DATE,                       "
                        +"     register_city        STRING,                  "
                        +"     register_job         STRING,                  "
                        +"     register_source_type INT,                     "
                        +"     gps_province STRING,                          "
                        +"     gps_city     STRING,                          "
                        +"     gps_region   STRING,                          "
                        +"     url_prefix    STRING,                         "
                        +"     page_type    STRING,                          "
                        +"     page_service STRING                           "
                        +" ) WITH (                                          "
                        +"  'connector' = 'kafka',                           "
                        +"  'topic' = 'dwd_events',                          "
                        +"  'properties.bootstrap.servers' = 'doitedu:9092', "
                        +"  'properties.group.id' = 'testGroup',             "
                        +"  'scan.startup.mode' = 'earliest-offset',         "
                        +"  'value.format'='json',                           "
                        +"  'value.json.fail-on-missing-field'='false',      "
                        +"  'value.fields-include' = 'EXCEPT_KEY')           "
        );


        // 6.建表，映射doris 中用于存储维度打宽明细数据的 表
        tenv.executeSql(
                " CREATE TABLE user_events_detail_doris (              "+
                        "   dt date,                                            "+
                        "   event_id string,                                    "+
                        "   user_id bigint,                                     "+
                        "   gps_province string,                                "+
                        "   gps_city string,                                    "+
                        "   gps_region string,                                  "+
                        "   username string,                                    "+
                        "   session_id string,                                  "+
                        "   event_time bigint,                                  "+
                        "   lat double,                                         "+
                        "   lng double,                                         "+
                        "   release_channel string,                             "+
                        "   device_type string,                                 "+
                        "   properties string,                                  "+
                        "   register_phone string,                              "+
                        "   user_status int,                                    "+
                        "   register_time timestamp(3),                         "+
                        "   register_gender int,                                "+
                        "   register_birthday date,                             "+
                        "   register_city string,                               "+
                        "   register_job string,                                "+
                        "   register_source_type int,                           "+
                        "   page_url string,                                    "+
                        "   url_prefix string,                                  "+
                        "   page_type string,                                   "+
                        "   page_service string                                 "+
                        " ) with (                                              "+
                        "     'connector' = 'doris',                            "+
                        "     'fenodes' = 'doitedu:8030',                       "+
                        "     'table.identifier' = 'dwd.user_events_detail',    "+
                        "     'username' = 'root',                              "+
                        "     'password' = 'root',                              "+
                        "     'sink.label-prefix' = 'doit41_3'                  "+
                        " )                                                     "
        );


        // 写一个维度关联的语句，并将关联结果创建为一个临时视图
        tenv.createTemporaryFunction("geocode", Gps2GeoHashUdf.class);
        tenv.executeSql(
                " CREATE TEMPORARY VIEW joined_view AS                                                                                                         "+
                        //" SELECT   "+
                        " SELECT  /*+ LOOKUP('table'='dim_user_info_hbase', 'retry-predicate'='lookup_miss', 'retry-strategy'='fixed_delay', 'fixed-delay'='5s','max-attempts'='3') */  "+
                        "     u.f1.id  as user_id                                                                                                                       "+
                        "     ,e.username                                                                                                                               "+
                        "     ,e.session_id                                                                                                                             "+
                        "     ,e.eventId as event_Id                                                                                                                    "+
                        "     ,e.eventTime as event_time                                                                                                                "+
                        "     ,e.lat                                                                                                                                    "+
                        "     ,e.lng                                                                                                                                    "+
                        "     ,e.release_channel                                                                                                                        "+
                        "     ,e.device_type                                                                                                                            "+
                        "     ,e.properties                                                                                                                             "+
                        "     ,u.f1.phone as register_phone                                                                                                             "+
                        "     ,u.f1.status as user_status                                                                                                               "+
                        "     ,u.f1.create_time as register_time                                                                                                        "+
                        "     ,u.f1.gender as register_gender                                                                                                           "+
                        "     ,u.f1.birthday as register_birthday                                                                                                       "+
                        "     ,u.f1.city as register_city                                                                                                               "+
                        "     ,u.f1.job as register_job                                                                                                                 "+
                        "     ,u.f1.source_type as register_source_type                                                                                                 "+
                        "     ,g.f.province as gps_province                                                                                                             "+
                        "     ,g.f.city as gps_city                                                                                                                     "+
                        "     ,g.f.region as gps_region                                                                                                                 "+
                        "     ,p.url_prefix                                                                                                                            "+
                        "     ,p.f.page_type                                                                                                                            "+
                        "     ,p.f.page_service                                                                                                                         "+
                        " from events_source e                                                                                                                          "+
                        " LEFT JOIN dim_user_info_hbase    FOR SYSTEM_TIME AS OF e.proc_time  AS  u  ON e.username=u.username                                           "+
                        " LEFT JOIN dim_geo_area_hbase     FOR SYSTEM_TIME AS OF e.proc_time  AS  g  ON geocode(e.lng,e.lat)=g.geohash                                  "+
                        " LEFT JOIN dim_page_info_hbase    FOR SYSTEM_TIME AS OF e.proc_time  AS  p  ON regexp_extract(e.properties['url'],'(^.*/).*?') = p.url_prefix  "
        );
        // tenv.executeSql("select * from joined_view").print();

        // 将关联结果写入kafka
        tenv.executeSql("insert into dwd_kafka select * from joined_view");


        // 将关联结果写入doris
        tenv.createTemporaryFunction("map2json", Map2JsonUdf.class);
        tenv.executeSql(
                " insert into user_events_detail_doris                                        "+
                        " select                                                                      "+
                        "     to_date(date_format(to_timestamp_ltz(event_time,3),'yyyy-MM-dd')) as dt "+
                        "     ,event_Id                                                               "+
                        "     ,user_id                                                                "+
                        "     ,gps_province                                                           "+
                        "     ,gps_city                                                               "+
                        "     ,gps_region                                                             "+
                        "     ,username                                                               "+
                        "     ,session_id                                                             "+
                        "     ,event_time                                                             "+
                        "     ,lat                                                                    "+
                        "     ,lng                                                                    "+
                        "     ,release_channel                                                        "+
                        "     ,device_type                                                            "+
                        "     ,map2json(properties) as properties                                     "+
                        "     ,register_phone                                                         "+
                        "     ,user_status                                                            "+
                        "     ,register_time                                                          "+
                        "     ,register_gender                                                        "+
                        "     ,register_birthday                                                      "+
                        "     ,register_city                                                          "+
                        "     ,register_job                                                           "+
                        "     ,register_source_type                                                   "+
                        "     ,properties['url'] as page_url                                          "+
                        "     ,url_prefix                                                             "+
                        "     ,page_type                                                              "+
                        "     ,page_service                                                           "+
                        " from joined_view                                                            "
        );
    }
}
